package com.qkjia.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
/**
 * @Auther 赵乾坤
 * @Date 2021-10-01 21:27
 * 统计发送消息成功和发送失败消息数，并在producer关闭时打印这两个计数器
 */
public class CounterInterceptor implements ProducerInterceptor<String, String>{

	// 失败个数
    private int errorCounter = 0;
    // 成功个数
    private int successCounter = 0;

	/**
	 * 配置
	 * @param configs
	 */
	@Override
	public void configure(Map<String, ?> configs) {
		
	}

	/**
	 * 发送时
	 * @param record
	 * @return
	 */
	@Override
	public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
		 return record;
	}

	/**
	 * 成功或失败个数
	 * @param metadata
	 * @param exception
	 */
	@Override
	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
		// 统计成功和失败的次数
        if (exception == null) {
            successCounter++;
        } else {
            errorCounter++;
        }
	}

	/**
	 * 关闭
	 */
	@Override
	public void close() {
        // 保存结果
        System.out.println("Successful sent: " + successCounter);
        System.out.println("Failed sent: " + errorCounter);
	}
}